package rabbitMQ-go

import (
	"fmt"
	"github.com/streadway/amqp"
	"log"
	"reflect"
	"strconv"
)

const MQURL = "amqp://guest:guest@127.0.0.1:5672/"


type RabbitMQ struct {
	conn    *amqp.Connection
	channel *amqp.Channel
	//队列名称
	queueName string
	//交换机名称
	exchangeName string
	//交换机类型
	exchangeKind string
	//bind Key 路由名称
	routeKey string
	//连接信息
	dsn string
	//交换机配置
	ExchangeArg amqp.Table
	//队列配置
	QueueArg amqp.Table
	//绑定配置
	QueueBindArg amqp.Table
	//消费者配置
	ConsumeArg amqp.Table
}

func (r *RabbitMQ) init() {
	var err error
	//获取Connection
	r.conn, err = amqp.Dial(r.dsn)
	r.failOnErr(err, "failed to Connect rabbitmq!")
	//获取Channel
	r.channel, err = r.conn.Channel()
	r.failOnErr(err, "failed to open a Channel")
}

//断开Channel 和 Connection
func (r *RabbitMQ) Destory() {
	_ = r.channel.Close()
	_ = r.conn.Close()
}

//设置队列
func (r *RabbitMQ) setQueue() {
	//1.申请队列，如果队列不存在会自动创建，存在则跳过创建
	q, err := r.channel.QueueDeclare(
		r.queueName,
		//是否持久化
		false,
		//是否自动删除
		false,
		//是否具有排他性
		false,
		//是否阻塞处理
		false,
		//额外的属性
		r.QueueArg,
	)
	r.queueName = q.Name
	if err != nil {
		fmt.Println("queue declare err" + err.Error())
	}
}

//
func (r *RabbitMQ) bindQueue() {
	err := r.channel.QueueBind(
		r.queueName,
		r.routeKey,
		r.exchangeName,
		false,
		r.QueueBindArg)
	if err != nil {
		fmt.Println("queue bind err" + err.Error())
	}
}

func (r *RabbitMQ) exchangeDeclare() {
	err := r.channel.ExchangeDeclare(
		r.exchangeName,
		r.exchangeKind,
		true,
		false,
		//true表示这个exchange不可以被client用来推送消息，仅用来进行exchange和exchange之间的绑定
		false,
		false,
		r.ExchangeArg, //,
	)
	r.failOnErr(err, "Failed to declare an exchange")
}

func (r *RabbitMQ) publish(message string) {
	err := r.channel.Publish(
		r.exchangeName,
		r.routeKey,
		false,
		false,
		amqp.Publishing{
			ContentType:  "text/plain",
			Body:         []byte(message),
			DeliveryMode: amqp.Persistent,
		})
	r.failOnErr(err, "Failed to publish message")
}

//推送消费模式
func (r *RabbitMQ) consume() <-chan amqp.Delivery {
	//接收消息
	msgs, err := r.channel.Consume(
		r.queueName, // queue
		//用来区分多个消费者
		"", // consumer
		//是否自动应答
		false, // auto-ack
		//是否独有
		false, // exclusive
		//设置为true，表示 不能将同一个Conenction中生产者发送的消息传递给这个Connection中 的消费者
		false, // no-local
		//列是否阻塞
		false,        // no-wait
		r.ConsumeArg, // args
	)
	if err != nil {
		fmt.Println(err)
	}
	return msgs
}

//拉取消费模式
func (r *RabbitMQ) get() amqp.Delivery {
	//接收消息
	msgs, ok, err := r.channel.Get(r.queueName, false)
	if !ok {
		fmt.Println("拉取信息失败")
	}
	if err != nil {
		fmt.Println(err)
	}
	return msgs
}

func (r *RabbitMQ) failOnErr(err error, msg string) {
	if err != nil {
		log.Fatalf("%s: %s", msg, err)
	}
}

func (r *RabbitMQ) Set(key string, value string) {
	t := reflect.ValueOf(r)
	field := t.Elem().FieldByName(key)
	switch field.Kind() {
	case reflect.Int64:
		i, _ := strconv.ParseInt(value, 10, 64)
		field.SetInt(i)
	case reflect.String:
		field.SetString(value)
	default:
		panic("设置类型暂不支持")
	}

}

type RabbitMQSimple struct {
	RabbitMQ
}

//创建简单模式下RabbitMQ实例
func NewRabbitMQSimple(dsn, queueName string) *RabbitMQSimple {
	//创建RabbitMQ实例
	RabbitMQSimple := &RabbitMQSimple{}
	RabbitMQSimple.dsn = dsn
	RabbitMQSimple.queueName = queueName
	RabbitMQSimple.init()
	return RabbitMQSimple
}

//直接模式队列生产
func (r *RabbitMQSimple) PublishSimple(message string) {
	//1.申请队列，如果队列不存在会自动创建，存在则跳过创建
	r.setQueue()
	//调用channel 发送消息到队列中
	r.routeKey = r.queueName //
	r.publish(message)
}

//simple 模式下 推送模式 消费者
func (r *RabbitMQSimple) RecieveSimple(fun func([]byte) bool) {
	//1.申请队列，如果队列不存在会自动创建，存在则跳过创建
	r.setQueue()
	//接收消息
	msgs := r.consume()

	forever := make(chan bool)
	//启用协程处理消息
	go func() {
		for d := range msgs {
			//消息逻辑处理，可以自行设计逻辑
			ok := fun(d.Body)
			if ok {
				d.Ack(false)
			}
		}
	}()

	log.Printf(" 退出请按 CTRL+C\n")
	<-forever
}

//simple 模式下 拉取模式 消费者
func (r *RabbitMQSimple) GetSimple(fun func([]byte) bool) {
	//1.申请队列，如果队列不存在会自动创建，存在则跳过创建
	r.setQueue()
	//接收消息
	msg := r.get()

	//启用协程处理消息
	go func() {
		//消息逻辑处理，可以自行设计逻辑
		ok := fun(msg.Body)
		if ok {
			msg.Ack(false)
		}
	}()

}

type RabbitMQPubSub struct {
	RabbitMQ
}

//订阅模式创建RabbitMQ实例
func NewRabbitMQPubSub(dsn, exchangeName string) *RabbitMQPubSub {
	//创建RabbitMQ实例
	RabbitMQPubSub := &RabbitMQPubSub{}
	RabbitMQPubSub.exchangeName = exchangeName
	RabbitMQPubSub.dsn = dsn
	RabbitMQPubSub.init()
	return RabbitMQPubSub
}

//订阅模式生产
func (r *RabbitMQPubSub) PublishPub(message string) {
	//1.尝试创建交换机
	r.exchangeKind = "fanout"
	r.exchangeDeclare()
	//2.发送消息
	r.publish(message)
}

//订阅模式消费端代码
func (r *RabbitMQPubSub) RecieveSub(fun func([]byte) bool) {
	//1.试探性创建交换机
	r.exchangeKind = "fanout"
	r.exchangeDeclare()
	//2.试探性创建队列，这里注意队列名称不要写
	r.setQueue()
	//绑定队列到 exchange 中
	r.bindQueue()
	//消费消息

	messges := r.consume()
	forever := make(chan bool)
	go func() {
		for d := range messges {
			ok := fun(d.Body)
			if ok {
				d.Ack(false)
			}
		}
	}()

	fmt.Println("退出请按 CTRL+C\n")
	<-forever
}

type RabbitMQRouting struct {
	RabbitMQ
}

//路由模式
//创建RabbitMQ实例
func NewRabbitMQRouting(dsn, exchangeName string) *RabbitMQRouting {
	//创建RabbitMQ实例
	RabbitMQRouting := &RabbitMQRouting{}
	RabbitMQRouting.exchangeName = exchangeName
	RabbitMQRouting.dsn = dsn
	RabbitMQRouting.init()
	return RabbitMQRouting
}

//路由模式发送消息
func (r *RabbitMQRouting) PublishRouting(routingKey string, message string) {
	//1.尝试创建交换机
	r.exchangeKind = "direct"
	r.exchangeDeclare()
	r.routeKey = routingKey
	//2.发送消息
	r.publish(message)
}

//路由模式接受消息
func (r *RabbitMQRouting) RecieveRouting(routingKey string, fun func([]byte) bool) {
	//1.试探性创建交换机
	r.exchangeKind = "direct"
	r.exchangeDeclare()
	//2.试探性创建队列，这里注意队列名称不要写
	r.setQueue()
	//绑定队列到 exchange 中
	r.routeKey = routingKey
	r.bindQueue()
	//消费消息
	messges := r.consume()
	forever := make(chan bool)

	go func() {
		for d := range messges {
			ok := fun(d.Body)
			if ok {
				d.Ack(false)
			}
		}
	}()
	fmt.Println("退出请按 CTRL+C\n")
	<-forever
}

//话题模式
type RabbitMQTopic struct {
	RabbitMQ
}

//创建RabbitMQ实例
func NewRabbitMQTopic(dsn, exchangeName string) *RabbitMQTopic {
	//创建RabbitMQ实例
	RabbitMQTopic := &RabbitMQTopic{}
	RabbitMQTopic.dsn = dsn
	RabbitMQTopic.exchangeName = exchangeName
	RabbitMQTopic.init()
	return RabbitMQTopic
}

//话题模式发送消息
func (r *RabbitMQTopic) PublishTopic(routingKey string, message string) {
	//1.尝试创建交换机
	r.exchangeKind = "topic"
	r.exchangeDeclare()
	r.routeKey = routingKey
	//2.发送消息
	r.publish(message)
}

//话题模式接受消息
//要注意routingKey规则
//其中“*”用于匹配一个单词，“#”用于匹配多个单词（可以是零个）
//匹配 myxy99.* 表示匹配 myxy99.hello, 但是myxy99.hello.one需要用myxy99.#才能匹配到
func (r *RabbitMQTopic) RecieveTopic(routingKey string, fun func([]byte) bool) {
	//1.试探性创建交换机
	r.exchangeKind = "topic"

	r.exchangeDeclare()
	//2.试探性创建队列，这里注意队列名称不要写
	r.setQueue()
	//绑定队列到 exchange 中
	r.routeKey = routingKey
	r.bindQueue()
	//消费消息
	messges := r.consume()

	forever := make(chan bool)
	go func() {
		for d := range messges {
			ok := fun(d.Body)
			if ok {
				d.Ack(false)
			}
		}
	}()

	fmt.Println("退出请按 CTRL+C\n")
	<-forever
}
